package com.cjh.wc;

import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

/**
 * @author chenjiahao
 * 无界流式处理，通过连接端口获取在另一设备上不断输入的数据进行处理
 */
public class StreamWordCount {

    public static void main(String[] args) throws Exception {
        // 1.创建流式执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 从参数中提取主机名和端口号
     /*   ParameterTool parameterTool = ParameterTool.fromArgs(args);
        String hostname = parameterTool.get("host");
        Integer port = parameterTool.getInt("port");*/

        //netcat安装模拟生产者发送数据到这里
        /**
         * 下载netcat。下载地址: https://eternallybored.org/misc/netcat/
         * 将nc.exe 复制到C:\Windows\System32的文件夹下
         * 打开cmd。输入nc 命令:nc -l -p 9000
         * 发送消息
         */
        // 2.读取文本流（无界流数据来源于源源不断，无法用文件模拟）
        DataStreamSource<String> lineDataStream = env.socketTextStream("127.0.0.1", 9000);

        // 3.转换计算
        SingleOutputStreamOperator<Tuple2<String, Long>> wordAndOneTuple = lineDataStream.flatMap((String line, Collector<Tuple2<String, Long>> out) -> {
            String[] words = line.split(" ");
            for (String word : words) {
                out.collect(Tuple2.of(word, 1L));
            }
        }).returns(Types.TUPLE(Types.STRING, Types.LONG));

        // 4.分组
        KeyedStream<Tuple2<String, Long>, String> wordAndOneKeyedStream = wordAndOneTuple.keyBy(data -> data.f0);

        // 5.求和
        SingleOutputStreamOperator<Tuple2<String, Long>> sum = wordAndOneKeyedStream.sum(1);

        // 6.输出
        sum.print();

        // 7.启动执行，流处理 只是定义流执行流程，需要启动流程，不停的等待数据的到来去输出
        env.execute();
    }
}
